<?php
/**
 * Created by PhpStorm.
 * User: bzg
 * Date: 2019/12/3
 * Time: 15:39
 */

namespace App\Extend\SysExtend\Queue;


use App\Extend\SysExtend\Cache\Cache;
use App\Extend\SysExtend\Helper;
use App\Extend\SysExtend\Queue\Driver\RabbitMQ;
use EasySwoole\Component\Di;
use EasySwoole\Component\Singleton;
use EasySwoole\Component\Timer;
use EasySwoole\EasySwoole\ServerManager;
use EasySwoole\Queue\Job;
use EasySwoole\Queue\Queue;

/**
 * 队列管理类
 * Class QueueManager
 * @package App\Extend\SysExtend\Queue
 */
class QueueManager
{
    use Singleton;

    /**
     * 消费者进程ID缓存前缀
     */
    const QUEUE_PID_PREFIX = 'queue_consumer_pid_';

    private $queue_list = [];


    private $config = [
        'driver' => RabbitMQ::class,
    ];


    /**
     * 注册队列  mainServerCreate事件中使用
     * @param array $queue [队列名，队列名=>消费者类全名]
     * @throws \Throwable
     */
    public function register(array $queue)
    {
        // 获取驱动类型
        if($config = Helper::config('queue')) {
            $this->config = $config;
        }
        // 将队列放入队列列表
        foreach ($queue as $k => $v) {
            $this->add($k, $v);
        }

        // 将自己放入容器
        Di::getInstance()->set('queueManager',$this);
    }


    /**
     * 添加队列到列表
     * @param array $queue
     * @throws \Exception
     */
    public function addQueue(array $queue)
    {
        // 将队列放入队列列表
        foreach ($queue as $k => $v) {
            $this->add($k, $v);
        }
    }

    /**
     * 从列表中移除队列
     * @param string ...$queue_names
     */
    public function delQueue(string ... $queue_names)
    {
        foreach ($queue_names as $n) {
            // 获取消费者进程ID
            $pid = $this->getPid($n);
            echo '结束进程',$pid;
            if($pid) {
                // 向进程发送终止信号
                \Swoole\Process::kill($pid,SIGTERM);
            }
            // 删除缓存的pid
            $this->rmPid($n);
            unset($this->queue_list[$n]);
        }
    }

    /**
     * 获取队列的消费者对象
     * @param string $name
     * @return mixed
     */
    public function getQueue(string $name)
    {
        return $this->queue_list[$name];
    }

    /**
     * 检查一个队列是否存在
     * @param string $name
     * @return bool
     */
    public function has(string $name)
    {
        return isset($this->queue_list[$name]);
    }

    /**
     * 往队列发送一条消息
     * @param string $name 队列名
     * @param mixed $data 信息
     * @return mixed
     */
    public function push(string $name, $data)
    {
        $job = new Job();
        $job->setJobData($data);
        return $this->queue_list[$name]->producer()->push($job);
    }

    /**
     * 往队列发送一条延时消息
     * @param string $name 队列名
     * @param mixed $data 数据
     * @param int $time 延时 秒
     */
    public function later(string $name, $data, int $time)
    {
        Timer::getInstance()->after($time * 1000,function () use ($name, $data){
            $job = new Job();
            $job->setJobData($data);
            $this->queue_list[$name]->producer()->push($job);
        });
    }

    /**
     * 设置消费者进程ID
     * @param string $queue_name
     * @param int $pid
     */
    public function setPid(string $queue_name, int $pid)
    {
        Cache::set(self::QUEUE_PID_PREFIX . $queue_name,$pid);
    }

    /**
     * 获取消费者进程ID
     * @param string $queue_name
     * @return mixed|null
     */
    public function getPid(string $queue_name)
    {
        return Cache::get(self::QUEUE_PID_PREFIX . $queue_name) ?? null;
    }

    /**
     * 移除PID
     * @param string $queue_name
     */
    public function rmPid(string $queue_name)
    {
        Cache::del(self::QUEUE_PID_PREFIX . $queue_name);
    }

    /**
     * 加入队列到列表，且启动消费进程
     * @param $k
     * @param $v
     * @throws \Exception
     */
    private function add($k, $v)
    {
        // 如果值是一个存在的类，则认为键为队列名，值为消费者类
        if(class_exists($v)) {
            $queue_name = $k;
            $consumer = $v;
        }
        // 否则，值为队列名，默认在Common/Queue目录下找大驼峰的队列名对应的类为消费者类
        else{
            $queue_name = $v;
            $consumer = 'App\Common\Queue\\' . \EasySwoole\Utility\Str::studly($v);
        }
        if(!isset($this->queue_list[$queue_name])) {
            $this->queue_list[$queue_name] = new Queue(new $this->config['driver']($queue_name, $this->config[$this->config['driver']]));
            // 注册一个消费进程
            ServerManager::getInstance()->addProcess(new $consumer());
        }
    }
}